package com.alibaba.otter.canal.deployer;

import java.util.Map;
import java.util.Properties;

import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.commons.lang.BooleanUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningData;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningListener;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors;
import com.alibaba.otter.canal.deployer.InstanceConfig.InstanceMode;
import com.alibaba.otter.canal.deployer.monitor.InstanceAction;
import com.alibaba.otter.canal.deployer.monitor.InstanceConfigMonitor;
import com.alibaba.otter.canal.deployer.monitor.ManagerInstanceConfigMonitor;
import com.alibaba.otter.canal.deployer.monitor.SpringInstanceConfigMonitor;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
import com.alibaba.otter.canal.instance.manager.PlainCanalInstanceGenerator;
import com.alibaba.otter.canal.instance.manager.plain.PlainCanalConfigClient;
import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator;
import com.alibaba.otter.canal.server.CanalMQStarter;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.exception.CanalServerException;
import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
import com.google.common.base.Function;
import com.google.common.collect.MapMaker;
import com.google.common.collect.MigrateMap;

/**
 * canal调度控制器（canal核心类）
 * 构造方法中，会对配置文件内容解析，初始化相关成员变量，
 * 做好canal server启动前的准备工作，之后在CanalStarter中调用CanalController.start()来启动
 *
 * @author jianghang 2012-11-8 下午12:03:11
 * @version 1.0.0
 */
public class CanalController {

    private static final Logger                      logger   = LoggerFactory.getLogger(CanalController.class);
    private String                                   ip;
    private String                                   registerIp;
    private int                                      port;
    private int                                      adminPort;
    // 默认使用spring的方式载入
    private Map<String, InstanceConfig>              instanceConfigs;
    // canal instance的全局配置,通过initGlobalConfig方法进行初始化
    private InstanceConfig                           globalInstanceConfig;
    private Map<String, PlainCanalConfigClient>      managerClients;
    // 监听instance config的变化
    private boolean                                  autoScan = true;
    private InstanceAction                           defaultAction;
    private Map<InstanceMode, InstanceConfigMonitor> instanceConfigMonitors;
    private CanalServerWithEmbedded                  embededCanalServer;
    private CanalServerWithNetty                     canalServer;
    // 该字段用于创建CanalInstance实例，是instance模块中的类
    private CanalInstanceGenerator                   instanceGenerator;
    private ZkClientx                                zkclientx;

    private CanalMQStarter                           canalMQStarter;
    private String                                   adminUser;
    private String                                   adminPasswd;

    public CanalController(){
        this(System.getProperties());
    }

    public CanalController(final Properties properties){
        managerClients = MigrateMap.makeComputingMap(new Function<String, PlainCanalConfigClient>() {

            public PlainCanalConfigClient apply(String managerAddress) {
                return getManagerClient(managerAddress);
            }
        });

        // 1.初始化全局参数设置
        globalInstanceConfig = initGlobalConfig(properties);
        // 利用Google Guava框架的MapMaker创建Map实例并赋值给instanceConfigs
        instanceConfigs = new MapMaker().makeMap();
        // 初始化instance config
        initInstanceConfig(properties);

        // init socketChannel
        String socketChannel = getProperty(properties, CanalConstants.CANAL_SOCKETCHANNEL);
        if (StringUtils.isNotEmpty(socketChannel)) {
            System.setProperty(CanalConstants.CANAL_SOCKETCHANNEL, socketChannel);
        }

        // 兼容1.1.0版本的ak/sk参数名
        String accesskey = getProperty(properties, "canal.instance.rds.accesskey");
        String secretkey = getProperty(properties, "canal.instance.rds.secretkey");
        if (StringUtils.isNotEmpty(accesskey)) {
            System.setProperty(CanalConstants.CANAL_ALIYUN_ACCESSKEY, accesskey);
        }
        if (StringUtils.isNotEmpty(secretkey)) {
            System.setProperty(CanalConstants.CANAL_ALIYUN_SECRETKEY, secretkey);
        }

        // 2.准备canal server
        // ip：String，对应canal.properties文件中的canal.ip，canal server监听的ip。
        // port：int，对应canal.properties文件中的canal.port，canal server监听的端口
        ip = getProperty(properties, CanalConstants.CANAL_IP);
        registerIp = getProperty(properties, CanalConstants.CANAL_REGISTER_IP);
        port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT, "11111"));
        adminPort = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_ADMIN_PORT, "11110"));
        // CanalServerWithEmbedded和CanalServerWithNetty都实现了CanalServer接口，
        // 且都实现了单例模式，通过静态方法instance获取实例。
        // server代表一个canal的运行实例，抽象了Embedded(嵌入式)/Netty(网络访问)两种实现
        // 可以不必独立部署canal server。在应用直接使用CanalServerWithEmbedded直连mysql数据库。
        // 如果技术hold不住相关代码，就独立部署一个canal server，使用canal提供的客户端，连接canal server获取binlog解析后数据。
        // 而CanalServerWithNetty是在CanalServerWithEmbedded的基础上做的一层封装，用于与客户端通信。
        // 因此，用于生成CanalInstance实例的instanceGenerator被设置到了CanalServerWithEmbedded中，
        // 而ip和port被设置到CanalServerWithNetty中。
        embededCanalServer = CanalServerWithEmbedded.instance();
        embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
        int metricsPort = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_METRICS_PULL_PORT, "11112"));
        embededCanalServer.setMetricsPort(metricsPort);

        this.adminUser = getProperty(properties, CanalConstants.CANAL_ADMIN_USER);
        this.adminPasswd = getProperty(properties, CanalConstants.CANAL_ADMIN_PASSWD);
        embededCanalServer.setUser(getProperty(properties, CanalConstants.CANAL_USER));
        embededCanalServer.setPasswd(getProperty(properties, CanalConstants.CANAL_PASSWD));

        String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY);
        if (canalWithoutNetty == null || "false".equals(canalWithoutNetty)) {
            // 将ip和port赋值给canalServer,类型为CanalServerWithNetty
            canalServer = CanalServerWithNetty.instance();
            canalServer.setIp(ip);
            canalServer.setPort(port);
        }

        // 处理下ip为空，默认使用hostIp暴露到zk中
        if (StringUtils.isEmpty(ip) && StringUtils.isEmpty(registerIp)) {
            ip = registerIp = AddressUtils.getHostIp();
        }
        if (StringUtils.isEmpty(ip)) {
            ip = AddressUtils.getHostIp();
        }
        if (StringUtils.isEmpty(registerIp)) {
            registerIp = ip; // 兼容以前配置
        }
        // canal支持利用了zk来完成HA机制、以及将当前消费到到的mysql的binlog位置记录到zk中。
        // ZkClientx是canal对ZkClient进行了一层简单的封装。
        // 3.初始化zk，读取canal.properties中的配置项canal.zkServers，
        final String zkServers = getProperty(properties, CanalConstants.CANAL_ZKSERVERS);
        // 如果没有这个配置，则表示项目不使用zk
        if (StringUtils.isNotEmpty(zkServers)) {
            zkclientx = ZkClientx.getZkClient(zkServers);
            // 初始化系统目录
            //destination列表，路径为/otter/canal/destinations
            zkclientx.createPersistent(ZookeeperPathUtils.DESTINATION_ROOT_NODE, true);
            //整个canal server的集群列表，路径为/otter/canal/cluster
            zkclientx.createPersistent(ZookeeperPathUtils.CANAL_CLUSTER_ROOT_NODE, true);
        }
        // 4.CanalInstance运行状态监控
        final ServerRunningData serverData = new ServerRunningData(registerIp + ":" + port);
        // ServerRunningMonitors是ServerRunningMonitor对象的容器，而ServerRunningMonitor用于监控CanalInstance。
        ServerRunningMonitors.setServerData(serverData);
        // 往ServerRunningMonitors设置Map时，是通过MigrateMap.makeComputingMap方法来创建的，
        // 其接受一个Function类型的参数，这是guava中定义的接口，其声明了apply抽象方法。
        ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
            // canal会为每一个destination创建一个CanalInstance，每个CanalInstance都会由一个ServerRunningMonitor来进行监控。
            // 而ServerRunningMonitor统一由ServerRunningMonitors进行管理。
            // 除了CanalInstance需要监控，CanalServer本身也需要监控。
            // 因此在代码一开始，就往ServerRunningMonitors设置了一个ServerRunningData对象，封装了canal server监听的ip和端口等信息。
            // 根据destination获取ServerRunningMonitor对象，如果不存在就创建（回调apply方法）。
            public ServerRunningMonitor apply(final String destination) {
                // 创建ServerRunningMonitor对象时，首先根据ServerRunningData创建ServerRunningMonitor实例，
                ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
                // 之后设置了destination和ServerRunningListener对象
                runningMonitor.setDestination(destination);
                // 启动一个CanalInstance实际上是在ServerRunningListener的processActiveEnter方法中，
                // 通过调用embededCanalServer的start(destination)方法进行的，对于停止也是类似。
                runningMonitor.setListener(new ServerRunningListener() {
                    /*内部调用了embededCanalServer的start(destination)方法。
    此处需要划重点，说明每个destination对应的CanalInstance是通过embededCanalServer的start方法启动的，
    这与之前分析将instanceGenerator设置到embededCanalServer中可以对应上。
    embededCanalServer负责调用instanceGenerator生成CanalInstance实例，并负责其启动。*/
                    public void processActiveEnter() {
                        try {
                            MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
                            embededCanalServer.start(destination);
                            if (canalMQStarter != null) {
                                canalMQStarter.startDestination(destination);
                            }
                        } finally {
                            MDC.remove(CanalConstants.MDC_DESTINATION);
                        }
                    }
                    // 内部调用embededCanalServer的stop(destination)方法。
                    // 与start方法类似，只不过是停止CanalInstance。
                    public void processActiveExit() {
                        try {
                            MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
                            if (canalMQStarter != null) {
                                canalMQStarter.stopDestination(destination);
                            }
                            embededCanalServer.stop(destination);
                        } finally {
                            MDC.remove(CanalConstants.MDC_DESTINATION);
                        }
                    }
                    /* 处理存在zk的情况下，在canal instance启动之前，在zk中创建节点。
                    路径为：/otter/canal/destinations/{0}/cluster/{1}，其0会被destination替换，1会被ip:port替换。
                    此方法会在processActiveEnter()之前被调用*/
                    public void processStart() {
                        try {
                            if (zkclientx != null) {
                                final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
                                    registerIp + ":" + port);
                                initCid(path);
                                zkclientx.subscribeStateChanges(new IZkStateListener() {
                                    public void handleStateChanged(KeeperState state) throws Exception { }
                                    public void handleNewSession() throws Exception {
                                        initCid(path);
                                    }

                                    @Override
                                    public void handleSessionEstablishmentError(Throwable error) throws Exception {
                                        logger.error("failed to connect to zookeeper", error);
                                    }
                                });
                            }
                        } finally {
                            MDC.remove(CanalConstants.MDC_DESTINATION);
                        }
                    }
                    // 处理存在zk的情况下，在Canalinstance停止前，释放zk节点，
                    // 路径为/otter/canal/destinations/{0}/cluster/{1}，
                    // 其0会被destination替换，1会被ip:port替换。此方法会在processActiveExit()之前被调用
                    public void processStop() {
                        try {
                            MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
                            if (zkclientx != null) {
                                final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
                                    registerIp + ":" + port);
                                releaseCid(path);
                            }
                        } finally {
                            MDC.remove(CanalConstants.MDC_DESTINATION);
                        }
                    }

                });
                // 接着，判断如果zkClientx字段如果不为空，也设置到ServerRunningMonitor中，
                if (zkclientx != null) {
                    runningMonitor.setZkClient(zkclientx);
                }
                // 最后调用init方法进行初始化。
                // 触发创建一下cid节点
                runningMonitor.init();
                return runningMonitor;
            }
        }));

        // 5.初始化monitor机制（autoScan机制）
        autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN));
        // 只有当autoScan为true时，才会初始化defaultAction字段和instanceConfigMonitors字段。
        if (autoScan) {
            // 其作用是如果配置发生了变更，默认应该采取什么样的操作。
            // 其实现了InstanceAction接口定义的三个抽象方法：start、stop和reload。
            defaultAction = new InstanceAction() {
                // 当新增一个destination配置时，需要调用start方法来启动；
                public void start(String destination) {
                    InstanceConfig config = instanceConfigs.get(destination);
                    if (config == null) {
                        // 重新读取一下instance config
                        config = parseInstanceConfig(properties, destination);
                        instanceConfigs.put(destination, config);
                    }

                    if (!embededCanalServer.isStart(destination)) {
                        // HA机制启动
                        ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                        if (!config.getLazy() && !runningMonitor.isStart()) {
                            runningMonitor.start();
                        }
                    }
                    // 这里也导致了虽然canal.properties中destinations项没有配置
                    // 但conf下有相应的配置文件，也会因为自动扫描机制，而启动这个instance
                    logger.info("auto notify start {} successful.", destination);
                }
                // 当移除一个destination配置时，需要调用stop方法来停止；
                public void stop(String destination) {
                    // 此处的stop，代表强制退出，非HA机制，所以需要退出HA的monitor和配置信息
                    InstanceConfig config = instanceConfigs.remove(destination);
                    if (config != null) {
                        embededCanalServer.stop(destination);
                        ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                        if (runningMonitor.isStart()) {
                            runningMonitor.stop();
                        }
                    }

                    logger.info("auto notify stop {} successful.", destination);
                }
                // 当某个destination配置发生变更时，需要调用reload方法来进行重启。
                public void reload(String destination) {
                    // 目前任何配置变化，直接重启，简单处理
                    stop(destination);
                    start(destination);

                    logger.info("auto notify reload {} successful.", destination);
                }

                @Override
                public void release(String destination) {
                    // 此处的release，代表强制释放，主要针对HA机制释放运行，让给其他机器抢占
                    InstanceConfig config = instanceConfigs.get(destination);
                    if (config != null) {
                        ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                        if (runningMonitor.isStart()) {
                            boolean release = runningMonitor.release();
                            if (!release) {
                                // 如果是单机模式,则直接清除配置
                                instanceConfigs.remove(destination);
                                // 停掉服务
                                runningMonitor.stop();
                                if (instanceConfigMonitors.containsKey(InstanceConfig.InstanceMode.MANAGER)) {
                                    ManagerInstanceConfigMonitor monitor = (ManagerInstanceConfigMonitor) instanceConfigMonitors.get(InstanceConfig.InstanceMode.MANAGER);
                                    Map<String, InstanceAction> instanceActions = monitor.getActions();
                                    if (instanceActions.containsKey(destination)) {
                                        // 清除内存中的autoScan cache
                                        monitor.release(destination);
                                    }
                                }
                            }
                        }
                    }

                    logger.info("auto notify release {} successful.", destination);
                }
            };
            // instanceConfigMonitors：类型为Map<InstanceMode, InstanceConfigMonitor>。
            // defaultAction字段只是定义了配置发生变化默认应该采取的操作，那么总该有一个类来监听配置是否发生了变化，
            // 这就是InstanceConfigMonitor的作用。
            // 官方文档中，只提到了对canal.conf.dir配置项指定的目录的监听，这指的是通过spring方式加载配置。
            // 显然通过manager方式加载配置，配置中心的内容也是可能发生变化的，也需要进行监听。
            // 因此instanceConfigMonitors的类型是一个Map，key为InstanceMode，就是为了对这两种方式的配置加载方式都进行监听。
            // instanceConfigMonitors也是根据mode属性，来采取不同的监控实现类
            // SpringInstanceConfigMonitor 或者ManagerInstanceConfigMonitor，
            // 二者都实现了InstanceConfigMonitor接口。
            instanceConfigMonitors = MigrateMap.makeComputingMap(new Function<InstanceMode, InstanceConfigMonitor>() {

                public InstanceConfigMonitor apply(InstanceMode mode) {
                    int scanInterval = Integer.valueOf(getProperty(properties,
                        CanalConstants.CANAL_AUTO_SCAN_INTERVAL,
                        "5"));

                    if (mode.isSpring()) {
                        //如果加载方式是spring，返回SpringInstanceConfigMonitor
                        SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor();
                        monitor.setScanIntervalInSecond(scanInterval);
                        monitor.setDefaultAction(defaultAction);
                        // 设置conf目录，默认是user.dir + conf目录组成
                        String rootDir = getProperty(properties, CanalConstants.CANAL_CONF_DIR);
                        if (StringUtils.isEmpty(rootDir)) {
                            rootDir = "../conf";
                        }

                        if (StringUtils.equals("otter-canal", System.getProperty("appName"))) {
                            monitor.setRootConf(rootDir);
                        } else {
                            // eclipse debug模式
                            monitor.setRootConf("src/main/resources/");
                        }
                        return monitor;
                    } else if (mode.isManager()) {
                        // 如果加载方式是manager，返回ManagerInstanceConfigMonitor
                        ManagerInstanceConfigMonitor monitor = new ManagerInstanceConfigMonitor();
                        monitor.setScanIntervalInSecond(scanInterval);
                        monitor.setDefaultAction(defaultAction);
                        String managerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER);
                        monitor.setConfigClient(getManagerClient(managerAddress));
                        return monitor;
                    } else {
                        throw new UnsupportedOperationException("unknow mode :" + mode + " for monitor");
                    }
                }
            });
        }
    }

    /**
     * 初始化canal instance的全局配置
     * 主要用于解析canal.properties以下几个配置项：
     *   canal.instance.global.mode：
     *     确定canal instance配置加载方式，取值有manager|spring两种方式
     *   canal.instance.global.lazy：
     *     确定canal instance是否延迟初始化
     * canal.instance.global.manager.address：配置中心地址。
     *   如果canal.instance.global.mode=manager，需要提供此配置项
     * canal.instance.global.spring.xml：spring配置文件路径。
     *   如果canal.instance.global.mode=spring，需要提供此配置项
     * 除了创建了globalInstanceConfig实例，同时还为字段instanceGenerator字段进行了赋值。
     */
    /**
     * canal.instance.global.mode用于确定canal instance的全局配置加载方式，其取值范围有2个：spring、manager。
     * 一个canal server中可以启动多个canal instance，每个instance都有各自的配置。
     * instance的配置也可以放在本地，也可以放在远程配置中心里。
     * 可以自定义每个canal instance配置文件存储的位置，
     * 也可以通过canal.instance.global.mode这个配置项，来统一的指定配置文件的位置，避免为每个canal instance单独指定。
     * spring方式：
     * 表示所有的canal instance的配置文件位于本地。
     * 此时必须提供配置项canal.instance.global.spring.xml指定spring配置文件的路径。
     * canal提供了多个spring配置文件（在类路径下的spring目录下）：
     * base-instance.xml、file-instance.xml、default-instance.xml、memory-instance.xml、group-instance.xml。
     * 这么多配置文件主要是为了支持canal instance不同的工作方式。
     * 而除了base-instance.xml本身，其他spring配置文件都在开头导入该配置文件：
     * <import resource="classpath:spring/base-instance.xml" />
     * manager方式：
     * 表示所有的canal instance的配置文件位于远程配置中心，
     * 此时必须提供配置项 canal.instance.global.manager.address来指定远程配置中心的地址。
     * 目前alibaba内部配置使用这种方式。开发者可以自己实现CanalConfigClient，连接各自的管理系统，完成接入。
     */
    private InstanceConfig initGlobalConfig(Properties properties) {
        String adminManagerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER);
        InstanceConfig globalConfig = new InstanceConfig();
        // 读取canal.instance.global.mode
        String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(CanalConstants.GLOBAL_NAME));
        if (StringUtils.isNotEmpty(adminManagerAddress)) {
            // 如果指定了admin manager地址,则强制适用manager
            globalConfig.setMode(InstanceMode.MANAGER);
        } else if (StringUtils.isNotEmpty(modeStr)) {
            // 将modelStr转成枚举InstanceMode，这是一个枚举类，只有2个取值，SPRING\MANAGER，对应两种配置方式
            globalConfig.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
        }
        // 读取canal.instance.global.lazy
        String lazyStr = getProperty(properties, CanalConstants.getInstancLazyKey(CanalConstants.GLOBAL_NAME));
        if (StringUtils.isNotEmpty(lazyStr)) {
            // 将字符串转换成boolean类型
            globalConfig.setLazy(Boolean.valueOf(lazyStr));
        }
        // 读取canal.instance.global.manager.address
        String managerAddress = getProperty(properties,
            CanalConstants.getInstanceManagerAddressKey(CanalConstants.GLOBAL_NAME));
        if (StringUtils.isNotEmpty(managerAddress)) {
            if (StringUtils.equals(managerAddress, "${canal.admin.manager}")) {
                managerAddress = adminManagerAddress;
            }

            globalConfig.setManagerAddress(managerAddress);
        }
        // 读取canal.instance.global.spring.xml
        String springXml = getProperty(properties, CanalConstants.getInstancSpringXmlKey(CanalConstants.GLOBAL_NAME));
        if (StringUtils.isNotEmpty(springXml)) {
            globalConfig.setSpringXml(springXml);
        }
        // 初始化instanceGenerator（通过匿名内部类）
        instanceGenerator = new CanalInstanceGenerator() {
            public CanalInstance generate(String destination) {
                // 1、根据destination从instanceConfigs获取对应的InstanceConfig对象
                // instanceConfigs 类型为 Map<String, InstanceConfig>
                InstanceConfig config = instanceConfigs.get(destination);
                if (config == null) {
                    throw new CanalServerException("can't find destination:" + destination);
                }
                //2、如果destination对应的InstanceConfig的mode是manager方式，使用PlainCanalInstanceGenerator
                if (config.getMode().isManager()) {
                    PlainCanalInstanceGenerator instanceGenerator = new PlainCanalInstanceGenerator(properties);
                    instanceGenerator.setCanalConfigClient(managerClients.get(config.getManagerAddress()));
                    instanceGenerator.setSpringXml(config.getSpringXml());
                    return instanceGenerator.generate(destination);
                //3、如果destination对应的InstanceConfig的mode是spring方式，使用SpringCanalInstanceGenerator
                } else if (config.getMode().isSpring()) {
                    // 创建了一个SpringCanalInstanceGenerator实例，然后往里面设置了spring xml
                    SpringCanalInstanceGenerator instanceGenerator = new SpringCanalInstanceGenerator();
                    // getSpringXml返回spring配置文件的路径
                    instanceGenerator.setSpringXml(config.getSpringXml());
                    // 通过generate方法获得CanalInstance实例
                    return instanceGenerator.generate(destination);
                } else {
                    throw new UnsupportedOperationException("unknow mode :" + config.getMode());
                }
            }
        };

        return globalConfig;
    }

    private PlainCanalConfigClient getManagerClient(String managerAddress) {
        return new PlainCanalConfigClient(managerAddress, this.adminUser, this.adminPasswd, this.registerIp, adminPort);
    }

    private void initInstanceConfig(Properties properties) {
        // 读取配置项canal.destinations
        String destinationStr = getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
        // 以","分割canal.destinations，得到一个数组形式的destination
        String[] destinations = StringUtils.split(destinationStr, CanalConstants.CANAL_DESTINATION_SPLIT);
        for (String destination : destinations) {
            // 为每一个destination生成一个InstanceConfig实例
            // 各个destination对应的InstanceConfig都是通过parseInstanceConfig方法来解析
            destination = destination.trim(); // 去掉首尾的空格
            InstanceConfig config = parseInstanceConfig(properties, destination);
            // 将destination对应的InstanceConfig放入instanceConfigs中
            InstanceConfig oldConfig = instanceConfigs.put(destination, config);

            if (oldConfig != null) {
                logger.warn("destination:{} old config:{} has replace by new config:{}", destination, oldConfig, config);
            }
        }
    }

    private InstanceConfig parseInstanceConfig(Properties properties, String destination) {
        String adminManagerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER);
        // 每个destination对应的InstanceConfig都引用了全局的globalInstanceConfig
        InstanceConfig config = new InstanceConfig(globalInstanceConfig);
        // 其他几个配置项与获取globalInstanceConfig类似，不再赘述，
        // 唯一注意的的是配置项的key部分中的global变成传递进来的destination
        String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(destination));
        if (StringUtils.isNotEmpty(adminManagerAddress)) {
            // 如果指定了manager地址,则强制适用manager
            config.setMode(InstanceMode.MANAGER);
        } else if (StringUtils.isNotEmpty(modeStr)) {
            config.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
        }

        String lazyStr = getProperty(properties, CanalConstants.getInstancLazyKey(destination));
        if (!StringUtils.isEmpty(lazyStr)) {
            config.setLazy(Boolean.valueOf(lazyStr));
        }

        if (config.getMode().isManager()) {
            String managerAddress = getProperty(properties, CanalConstants.getInstanceManagerAddressKey(destination));
            if (StringUtils.isNotEmpty(managerAddress)) {
                if (StringUtils.equals(managerAddress, "${canal.admin.manager}")) {
                    managerAddress = adminManagerAddress;
                }
                config.setManagerAddress(managerAddress);
            }
        } else if (config.getMode().isSpring()) {
            String springXml = getProperty(properties, CanalConstants.getInstancSpringXmlKey(destination));
            if (StringUtils.isNotEmpty(springXml)) {
                config.setSpringXml(springXml);
            }
        }

        return config;
    }

    public static String getProperty(Properties properties, String key, String defaultValue) {
        String value = getProperty(properties, key);
        if (StringUtils.isEmpty(value)) {
            return defaultValue;
        } else {
            return value;
        }
    }

    public static String getProperty(Properties properties, String key) {
        key = StringUtils.trim(key);
        String value = System.getProperty(key);

        if (value == null) {
            value = System.getenv(key);
        }

        if (value == null) {
            value = properties.getProperty(key);
        }

        return StringUtils.trim(value);
    }
    // ServerRunningMonitor的start方法，是在CanalController中的start方法中被调用的，
    // 而CanalController中的start方法是在CanalLauncher中被调用的。
    public void start() throws Throwable {
        logger.info("## start the canal server[{}({}):{}]", ip, registerIp, port);
        // 创建整个canal的工作节点 :/otter/canal/cluster/{0}
        final String path = ZookeeperPathUtils.getCanalClusterNode(registerIp + ":" + port);
        initCid(path);
        if (zkclientx != null) {
            this.zkclientx.subscribeStateChanges(new IZkStateListener() {

                public void handleStateChanged(KeeperState state) throws Exception { }

                public void handleNewSession() throws Exception {
                    initCid(path);
                }

                @Override
                public void handleSessionEstablishmentError(Throwable error) throws Exception {
                    logger.error("failed to connect to zookeeper", error);
                }
            });
        }
        // 优先启动embeded服务
        embededCanalServer.start();
        // 尝试启动一下非lazy状态的通道
        // 通过迭代instanceConfigs，根据destination获取对应的ServerRunningMonitor，然后逐一启动
        for (Map.Entry<String, InstanceConfig> entry : instanceConfigs.entrySet()) {
            final String destination = entry.getKey();
            InstanceConfig config = entry.getValue();
            // 创建destination的工作节点
            if (!embededCanalServer.isStart(destination)) {
                // HA机制启动
                ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                // 如果不是lazy，lazy模式需要等到第一次有客户端请求才会启动
                if (!config.getLazy() && !runningMonitor.isStart()) {
                    runningMonitor.start();
                }
            }

            if (autoScan) {
                instanceConfigMonitors.get(config.getMode()).register(destination, defaultAction);
            }
        }

        if (autoScan) {// 启动配置文件自动检测机制
            instanceConfigMonitors.get(globalInstanceConfig.getMode()).start();
            for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) {
                if (!monitor.isStart()) {
                    monitor.start(); // 启动monitor
                }
            }
        }

        // 启动网络接口，监听客户端请求
        if (canalServer != null) {
            canalServer.start();
        }
    }

    public void stop() throws Throwable {

        if (canalServer != null) {
            canalServer.stop();
        }

        if (autoScan) {
            for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) {
                if (monitor.isStart()) {
                    monitor.stop();
                }
            }
        }

        for (ServerRunningMonitor runningMonitor : ServerRunningMonitors.getRunningMonitors().values()) {
            if (runningMonitor.isStart()) {
                runningMonitor.stop();
            }
        }

        // 释放canal的工作节点
        releaseCid(ZookeeperPathUtils.getCanalClusterNode(registerIp + ":" + port));
        logger.info("## stop the canal server[{}({}):{}]", ip, registerIp, port);

        if (zkclientx != null) {
            zkclientx.close();
        }

        // 关闭时清理缓存
        if (instanceConfigs != null) {
            instanceConfigs.clear();
        }
        if (managerClients != null) {
            managerClients.clear();
        }
        if (instanceConfigMonitors != null) {
            instanceConfigMonitors.clear();
        }

        ZkClientx.clearClients();
    }

    private void initCid(String path) {
        // logger.info("## init the canalId = {}", cid);
        // 初始化系统目录，这里的节点是ephemeral
        if (zkclientx != null) {
            try {
                zkclientx.createEphemeral(path);
            } catch (ZkNoNodeException e) {
                // 如果父目录不存在，则创建
                String parentDir = path.substring(0, path.lastIndexOf('/'));
                zkclientx.createPersistent(parentDir, true);
                zkclientx.createEphemeral(path);
            } catch (ZkNodeExistsException e) {
                // ignore
                // 因为第一次启动时创建了cid,但在stop/start的时可能会关闭和新建,允许出现NodeExists问题s
            }

        }
    }

    private void releaseCid(String path) {
        // logger.info("## release the canalId = {}", cid);
        // 初始化系统目录
        if (zkclientx != null) {
            zkclientx.delete(path);
        }
    }

    public CanalMQStarter getCanalMQStarter() {
        return canalMQStarter;
    }

    public void setCanalMQStarter(CanalMQStarter canalMQStarter) {
        this.canalMQStarter = canalMQStarter;
    }

    public Map<InstanceMode, InstanceConfigMonitor> getInstanceConfigMonitors() {
        return instanceConfigMonitors;
    }

    public Map<String, InstanceConfig> getInstanceConfigs() {
        return instanceConfigs;
    }

}
